package com.tt.vtg.scenario.http.handler;

import com.google.common.collect.ImmutableMap;
import com.tt.vtg.scenario.http.HttpJsonRequest;
import com.tt.vtg.scenario.http.HttpJsonResponse;
import com.tt.vtg.scenario.kafka.KafkaInboundMsg;
import com.tt.vtg.scenario.kafka.KafkaProducerTemplate;
import com.tt.vtg.scenario.redis.JedisTemplate;
import com.tt.vtg.scenario.task.CheckOrderEntity;
import com.tt.vtg.scenario.http.pojo.RequestEntity;
import com.tt.vtg.scenario.http.pojo.ResponseEntity;
import com.tt.vtg.scenario.task.CheckOrderStatusTask;
import com.tt.vtg.scenario.util.json.JsonUtils;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;

import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;

public class HttpJsonServerHandler extends SimpleChannelInboundHandler<HttpJsonRequest<RequestEntity>> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, HttpJsonRequest<RequestEntity> msg) throws Exception {
        FullHttpRequest request = msg.getRequest();
        RequestEntity requestEntity = msg.getObj();

        ResponseEntity responseEntity = null;
        FullHttpResponse response = null;

        CheckOrderEntity orderChecked = process(requestEntity);

        if(orderChecked == null) {
            sendError(ctx);
            return;
        }

        switch (orderChecked.getStatus()) {
            case CheckOrderStatusTask.SUCCESS: {
                responseEntity = ResponseEntity.SUCC();
                responseEntity.setResponseMsg(orderChecked.getResult());
                response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
                break;
            }
            case CheckOrderStatusTask.PROCESSING:
            case CheckOrderStatusTask.FAILED: {
                responseEntity = ResponseEntity.FAILED();
                responseEntity.setResponseMsg(
                        orderChecked.getResult()!=null ?
                                orderChecked.getResult() :
                                "The operation is unsuccessful, pls try again later. ");

                response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR);

                break;
            }
        }

        ChannelFuture future = ctx.writeAndFlush(new HttpJsonResponse<>(response, responseEntity));

        // check whether the 'Connection: Keep-Alive' is setting.
        //http 1.1 is setting by default.
        if(!HttpUtil.isKeepAlive(request)){
            future.addListener(ChannelFutureListener.CLOSE);
        }
    }

    private CheckOrderEntity process(RequestEntity requestEntity) {
        String orderId = genRecordId();
        // firstly set order into redis
        persistOrder(orderId, requestEntity);

        // secondly send msg to kafka
        sendOrder(orderId, requestEntity);

        // lastly check order status
        return getOrderStatus(orderId);
    }

//    private CheckOrderEntity getOrderStatus(String orderId) {
//        FutureTask<CheckOrderEntity> f = new FutureTask<>(new CheckOrderStatusTask(orderId));
//        f.run();
//        try {
//            return f.get(1, TimeUnit.SECONDS);
//        } catch (InterruptedException | ExecutionException | TimeoutException e) {
//            e.printStackTrace();
//        }
//
//        return null;
//    }

    private CheckOrderEntity getOrderStatus(String orderId) {
        CheckOrderEntity checkedEntity = new CheckOrderEntity();
        int tryTime=0;
        for(;;){
            ++tryTime;

            JedisTemplate.execute(j->{
                List<String> lst = j.hmget(orderId, "status", "result");
                checkedEntity.setStatus(Integer.valueOf((lst.get(0)==null ? String.valueOf(CheckOrderStatusTask.FAILED) : lst.get(0))));
                checkedEntity.setResult(lst.get(1));

                return null;
            });

            if(checkedEntity.getStatus() != CheckOrderStatusTask.PROCESSING || tryTime >= 5){
                return checkedEntity;
            }

            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private void sendOrder(String orderId, RequestEntity requestEntity) {
        KafkaInboundMsg inboundRecord= new KafkaInboundMsg();
        inboundRecord.setOrderId(orderId);
        inboundRecord.setPayload(JsonUtils.toJson(requestEntity));

        KafkaProducerTemplate.put(inboundRecord);
    }

    private void persistOrder(String orderId, RequestEntity requestEntity){
        JedisTemplate.execute(jedis -> {
            Map<String, String> order = ImmutableMap.<String, String>builder().
                    put("userId", requestEntity.getUserId()).
                    put("action", requestEntity.getAction().name()).
                    put("description", requestEntity.getAction().name() + " " + requestEntity.getAmount()).
                    put("amount", String.valueOf(requestEntity.getAmount())).
                    put("status", String.valueOf(CheckOrderStatusTask.PROCESSING)).build();

            jedis.hmset(orderId, order);
            // expire after 60 second.
            jedis.expire(orderId, 60);

            return null;
        });
    }

    private String genRecordId(){
        return UUID.randomUUID().toString();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("HttpJsonServerHandler @ exceptionCaught");
        cause.printStackTrace();
        if(ctx.channel().isActive()){
            sendError(ctx);
        }
    }

    private void sendError(ChannelHandlerContext ctx) {
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                HttpResponseStatus.INTERNAL_SERVER_ERROR,
                Unpooled.copiedBuffer("Internal server error, please check log in server...", CharsetUtil.UTF_8));

        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");

        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
    }
}
